package reactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.SynchronousSink;
import reactor.core.scheduler.Schedulers;
import util.Util;

import java.util.concurrent.CountDownLatch;
import java.util.function.BiFunction;

/**
 * @author yding5
 */
public class ParallelTest {

    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        Flux.generate(() -> 0, (BiFunction<Integer, SynchronousSink<Integer>, Integer>)(state, sink) -> {
            if (state == 10) {
                // state更新至5时，结束生产
                sink.complete();
            } else {
                // 生产内容
                Util.print("发射了元素：" + state);
                sink.next(state);
            }
            return ++state;
        }).parallel(3, 6)
            .runOn(Schedulers.elastic(), 2)
            .subscribe(
                    // onNext
                    (x) -> Util.print("消费了元素" + x),
                    // onError
                    Throwable::printStackTrace,
                    // onComplete
                    () -> {
                        Util.print("Subscriber consume finished.");
                        countDownLatch.countDown();
                    });
        countDownLatch.await();
    }
}
